from fastapi import FastAPI

app = FastAPI()

# app.dependency_overrides


# openapi 它提供了一种描述 API 结构、端点、请求参数、响应数据等的通用语言，使得开发者可以更容易地理解和使用 API。


# 请详细介绍一下FastAPI是如何处理HTTP请求的

# 浏览器 --- asgi服务器解析http请求转成asgi格式的 scope ---  经过中间件处理 ---- fastapi路由匹配，-----  执行 get方法
# ----  依赖注入 ---- 构建上下文环境 ----  递归执行依赖 ---- 返回结果 --- 参数处理（数据验证相关） --- 执行路径函数
# ----- orm 相关操作 ------ 生成相应 ---- 中间件 --- asgi 返回给客户端





# FastAPI的依赖注入非常强大，可以用于共享业务逻辑、权限验证、数据库会话等。
# 对于不需要在响应前完成的操作，可以使用后台任务（例如发送邮件、处理文件等）
# FastAPI内置了OAuth2密码流和JWT的支持，适用于用户认证。
"""
    使用 uname  password  --->  数据查询  ---->  如果正确创建token：create_access_token(data={"sub": user.username})
    ----->  token 的创建使用 jwt：jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
"""
# 使用APIRouter进行模块化：当应用变大时，使用APIRouter将路由分组到不同的模块中
# FastAPI使用Depends函数来声明依赖，支持函数、类和生成器三种形式
# 依赖注入的核心原理包括类型提示检查、依赖图构建和结果传递



# ---------------------------Depends 依赖注入-------------------------------------------
# 设计兼顾了灵活性与性能  FastAPI通过Depends()声明构建有向无环依赖树


# Depends() 利用 Python 的类型注解（如 Session = Depends(get_db)）和 FastAPI 的请求上下文，自动解析参数类型。
#                   依赖项函数的参数可通过路径参数、查询参数、请求头等自动填充，与路由函数使用相同的参数解析逻辑
#                   依赖执行顺序​​：依赖项按声明顺序递归解析
# 依赖关系管理​​：FastAPI 通过 solve_dependencies() 函数将依赖树展平为执行序列，确保无循环引用

# 请求级缓存​​：默认开启 (use_cache=True)。同一请求中多次声明相同依赖（如多个路由函数依赖同一个数据库连接）时，
#           仅首次执行依赖函数，后续直接复用缓存结果。
    # ​​作用域控制​​：缓存生命周期绑定到单次请求，请求结束后自动清除。
    #           对于需全局缓存的依赖（如配置加载），可结合单例模式实现应用级缓存。

# 优势​​：显著减少高开销依赖的重复执行（如数据库连接、认证检查），降低延迟。
#         实测显示，对复杂依赖树开启缓存可提升 30%-50% 的吞吐量   
#         若依赖需实时数据（如实时权限验证），需设置 use_cache=False 禁用缓存。
#         滥用全局缓存可能导致内存泄漏或状态污染，建议仅缓存无状态或只读依赖


# ​​层级扁平化​​：避免超过 5 层的嵌套依赖，减少解析深度。
# ​​异步依赖优化​​：异步依赖（async def）可并行执行 I/O 操作，但需确保子依赖兼容异步（否则阻塞事件循环
# ​​场景​​	​​use_cache 设置​​	​​理由​
# 数据库连接池	True (默认)	避免重复建立连接
# 实时用户权限验证	False	确保每次请求检查最新权限
# 配置加载	应用级单例	避免重复读取文件/环境变量

# Depends() 通过类型注解自动解析依赖，构建树形结构实现层级注入。其缓存机制是性能关键：
# ✅ ​​适用缓存场景​​：无状态、高开销操作（DB 连接、配置加载）。
# ❌ ​​禁用缓存场景​​：需实时响应的操作（权限验证、动态配置）。
# 最佳实践包括控制依赖层级深度、异步化 I/O 操作、定期审计依赖树复杂度。
#               这些策略可确保依赖注入在提供解耦和复用能力的同时，维持高性能服务。





# ---------------------------  依赖覆盖  -------------------------------------------
# 通过 app.dependency_overrides 字典临时替换依赖函数，无需修改原代码
# 生产环境依赖
def real_db():
    return RealDatabase()

# 测试环境模拟
def mock_db():
    return MockDatabase()

# 测试前覆盖
app.dependency_overrides[real_db] = mock_db

# 测试结束后调用 app.dependency_overrides.clear() 恢复原始依赖


# 测试环境集成​​
# 结合 pytest 的 Fixture 自动管理依赖生命周期：
# conftest.py
@pytest.fixture
def test_client():
    app.dependency_overrides[get_db] = lambda: "sqlite:///:memory:"
    with TestClient(app) as client:
        yield client
    app.dependency_overrides.clear()  # 测试后清理



# 依赖解析优先级​​: FastAPI 在处理请求时，优先检查 dependency_overrides 字典。若存在覆盖项，则调用覆盖函数而非原始依赖，形成运行时动态替换

# 作用域控制​​: 依赖覆盖默认作用于整个应用生命周期。若需限制范围（如单次请求），可通过生成器依赖结合 yield 实现资源清理
@asynccontextmanager
async def scoped_dependency():
    resource = init_resource()
    yield resource
    resource.cleanup()  # 请求结束后清理


        # 通过 __fastapi_scope__ 属性可自定义作用域（如 "app" 或 "function"）       


# 最佳实践​
# 类型一致性​​：确保覆盖函数返回值类型与原依赖一致，避免因类型错误引发Pydantic校验失败（如返回字典替代模型对象）。
# ​​依赖隔离​​：将外部服务（数据库、API）封装为独立依赖，便于替换。
# ​​覆盖率统计​​：配置 pytest-cov 包含依赖模块（如 --cov=app.dependencies），避免遗漏关键逻


# 💎 ​​总结​​
# FastAPI 的依赖覆盖机制通过 ​​动态重写依赖字典​​ 和 ​​依赖图重构​​ 实现环境隔离，核心价值在于：

# ​​解耦性​​：业务逻辑与基础设施分离，通过覆盖注入模拟实现无侵入式测试。
# ​​灵活性​​：支持嵌套依赖替换，适应复杂场景（如数据库连接+权限链）。
# ​​可维护性​​：结合 pytest 自动化管理依赖生命周期，降低测试维护成本






# ---------------------------  实操  -------------------------------------------
# 1. 构建一个请求调用关系的有向无环，树结构，自底向上依次调用
# 2. 依赖缓存，同一个调用只会调用一次

# 3. 调用时机 包括： 直接调用 和间接调用（好多个依赖项串起来）

# 4. 生命周期管理：
# # yield前代码在依赖解析阶段执行，finally在响应返回后执行。
# def get_db():
#     db = SessionLocal()   # 请求开始时创建连接
#     try: yield db
#     finally: db.close()   # 请求结束后自动清理[1](@ref)


# # 异步上下文管理器：对异步资源（如Redis连接池），使用@asynccontextmanager
# from contextlib import asynccontextmanager

# @asynccontextmanager
# async def get_redis():
#     pool = await create_redis_pool()
#     try: yield pool
#     finally: await pool.close()  # 异步清理[3](@ref)


# 企业实践：
# 1. ​​分层权限验证​：​	认证依赖(get_user) → 角色验证依赖(RoleChecker) → 路由依赖	   ---------------------  权限逻辑复用，支持动态角色配置
# 2. 多数据库事务​：​	主从库分离：@app.middleware("http")中根据请求类型选择注入读/写库依赖	   ---------------------  读写分离负载均衡
# 3. 请求级缓存​：​	在依赖中初始化缓存对象，通过request.state.cache跨依赖共享	   ---------------------  避免重复计算，提升性能
# 4. ​​动态配置加载：​​	应用启动时加载配置→依赖中通过app.state.config访问	   ---------------------  避免频繁IO，配置热更新无需重启服务



# ---------------------------   注意事项：  ------------------- 
# ​​循环依赖预防
# 资源泄漏风险：生成器依赖中必须用try/finally确保资源释放； 异步依赖清理需等待finally中的异步操作完成
# 作用域一致性​​：数据库连接等有状态资源必须设为request作用域，禁止跨请求共享
#         # 错误示范：应用级数据库连接
#         db = SessionLocal()  # 多请求竞争导致数据错乱

#         # 正确方案：每个请求独立连接
#         def get_db(): yield SessionLocal()


# 💎💎 掌握这些机制可设计出高内聚低耦合的服务模块，尤其在需要精细控制数据库事务或分布式权限链的场景中体现核心价值 💎💎





# ---------------------------   参数解析优先级与覆盖   ------------------- 
# 路径参数（Path Parameters）​​
# URL 路径中的占位符（如 /users/{id}）优先级最高，框架首先提取并验证类型（如 id: int）。若类型不匹配，直接返回 422 错误
@app.get("/users/{user_id}")
def get_user(user_id: int): ...  # 优先解析 user_id

# ​​查询参数（Query Parameters
        # /users/admin?is_admin=true   ----------->  role="admin"（路径参数），is_admin=True（查询参数）
@app.get("/users/{role}")
def get_users(role: str, is_admin: bool = False): ...



# 参数覆盖场景与机制​

# 当多个路由路径相似时，​​声明顺序决定匹配优先级​​
@app.get("/files/{file_path}")   # 优先匹配
def read_file(file_path: str): ...

@app.get("/files/static")        # 无法匹配 /files/static
def static_file(): ...

        # --- 解决方案​​---
            # 1. 将精确路径（如 /files/static）放在模糊路径（如 /files/{file_path}）之前
            # 2. 使用 Path(..., regex="^static$") 约束路径参数



# 查询参数覆盖依赖项默认值​​  ----  /data?limit=20
def dep_params(limit: int = 10):   # ---- limit 是 url传过来的
    return limit

@app.get("/data")
def get_data(limit: int = Depends(dep_params)):   # ------------ limit 是 依赖函数返回的
    return {"limit": limit}


# FastAPI 通过 ​​路径参数 → 查询参数 → 依赖项 → 请求体​​ 的优先级链，结合 ​​声明顺序、类型强制、依赖覆盖​​ 三大机制，实现了灵活且安全的参数解析
# 💎 路径参数定位资源，查询参数控制行为，依赖项封装共享逻辑 💎




# --------------------------------- asgi -----------------------------------------------
# ASGI 是 Python 的异步 Web 服务器标准，通过三个核心参数驱动请求处理
# ​​scope 作用域字典​：存储请求的元数据（如 URL、方法、路径参数、协议类型），生命周期仅限单次请求。例如，HTTP 请求的 scope 包含 "type": "http"，WebSocket 则为 "websocket"
# ​​receive 异步接收函数​：用于从客户端分段读取数据流（如大文件上传），返回结构化的消息字典（如 {"type": "http.request", "body": b"data", "more_body": True}）
# send 异步发送函数​：向客户端返回响应（如 {"type": "http.response.start", "status": 200} 和 {"type": "http.response.body", "body": b"Hello"}）。

# ​​FastAPI 的封装​​：Starlette 将原始 ASGI 消息转化为高层对象（如 Request），开发者无需直接操作底层消息格式。



# Request 对象的本质：ASGI 消息的抽象
# FastAPI 的 Request 对象（继承自 Starlette）是对 ASGI 作用域和消息的抽象封装
# ​​数据访问接口​​：
#     ​​路径/查询参数​​：通过 scope 中的 path_params 和 query_string 解析生成。
#     ​​请求头​​：从 scope["headers"] 提取并解码为字典。
#     ​​请求体​​：调用 await request.json() 触发 receive 消息读取，并自动解析 JSON。
# ​​生命周期扩展​​：
#     request.state 属性允许在中间件和路由间传递自定义数据（如数据库连接）
from fastapi import Request

@app.get("/items/{id}")
async def read_item(request: Request, id: int):
    client_ip = request.client.host  # 从 scope 提取客户端 IP
    user_agent = request.headers.get("user-agent")  # 解码 headers
    return {"id": id, "ip": client_ip}


# Response 对象负责将业务逻辑输出转换为 ASGI 消息
# ​​自动序列化：返回 Python 字典时，FastAPI 使用 Pydantic 模型将其转为 JSON，并通过 send 发送 HTTP 响应消息
# ​​流式响应支持​​：对大文件或流数据，通过异步生成器逐步发送 http.response.body 消息（避免内存溢出）
async def stream_data():
    yield b"Part1"
    yield b"Part2"
    return StreamingResponse(stream_data())



# FastAPI 基于 Starlette 的 ASGI 基础实现关键性能设计
# ​​异步路由函数​​：使用 async def 定义路由，在 I/O 等待时释放事件循环（如数据库查询），提升并发能力。
# ​​依赖注入系统​​：依赖项可异步执行（如认证检查），通过 ASGI 消息队列协调执行顺序。
# ​​中间件机制​​：基于 ASGI 的中间件可拦截请求/响应消息（如 CORS 处理），修改 scope 或响应头


# 💎 总结
# ASGI 协议统一了服务器与框架交互方式，使 FastAPI 可运行于任何 ASGI 服务器（如 Uvicorn）
# 步消息模型支撑高并发（尤其 I/O 密集型场景），WebSocket/HTTP/2 等协议可无缝扩展
# 隐藏底层消息复杂性，暴露简洁的 Request/Response 接口，结合 Pydantic 实现类型安全

#  Request 和 Response 对象本质是 ASGI 作用域与消息的抽象，将底层协议细节转化为开发者友好的高级接口






# --------------------------------- 中间件 -----------------------------------------------
FastAPI 的中间件机制基于 Starlette 的 ASGI 实现，通过拦截请求/响应流程实现跨切面功能


# 中间件按注册顺序​​反向嵌套​​执行，形成“洋葱式”处理流程
app.add_middleware(A)  # 外层
app.add_middleware(B)  # 内层
# 执行顺序：A请求前 → B请求前 → 路由 → B响应后 → A响应后

@app.middleware("http")
async def outer(request, call_next):
    print("outer-before")    # 1. 最先执行
    response = await call_next(request)  # 调用内层
    print("outer-after")     # 5. 最后执行
    return response

@app.middleware("http")
async def inner(request, call_next):
    print("inner-before")    # 2. 次之执行
    response = await call_next(request)  # 调用路由
    print("inner-after")     # 4. 次后执行
    return response

# ​​输出顺序​​：outer-before → inner-before → 路由逻辑 → inner-after → outer-after


# 异常关键点​​：
# 内层中间件或路由的异常会​​逆序传递​​到外层中间件。
# 认证类中间件（如API密钥校验）应在外层注册，确保非法请求被提前拦截





# call_next 的核心作用：  1. 连接中间件与路由；  2. 异步调度关键；  3. ​​修改响应入口
# 负责将请求传递给下一个中间件或最终路由函数
# 当调用 await call_next(request) 时，系统会暂停当前中间件，执行后续处理流程，返回后再继续当前逻辑
                            #   释放事件循环，允许其他请求被处理，避免阻塞      
async def middleware(request, call_next):
    # I/O 等待期间事件循环可处理其他任务
    response = await call_next(request)  # ⚡ 非阻塞点
    return response

# 获取路由生成的响应对象后，可修改其内容、状态码或头部
response = await call_next(request)
response.headers["X-Data-Source"] = "Database"


# 中间件---应用（全局）
# 修改请求/响应头	
# CORS
# 认证逻辑
# 高并发日志记录	
# 全局异常捕获
# 限流




# 自定义中间件实现
@app.middleware("http")    #---------------​​修改请求/响应头​
async def modify_headers(request: Request, call_next):
    # 请求前：添加自定义请求头
    request.headers.update({"X-Request-ID": "123"})
    
    response = await call_next(request)
    
    # 响应后：添加性能标记
    response.headers["X-Response-Time"] = "100ms"
    return response



@app.middleware("http")    #---------------​​​CORS 处理
async def cors_middleware(request: Request, call_next):
    response = await call_next(request)
    response.headers["Access-Control-Allow-Origin"] = "*"
    response.headers["Access-Control-Allow-Methods"] = "GET,POST"
    return response


#---------------官方中间件
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["*"]
)




@app.middleware("http")     #---------------认证逻辑​
async def auth_middleware(request: Request, call_next):
    api_key = request.headers.get("X-API-Key")
    if api_key != "SECRET_KEY":
        raise HTTPException(
            status_code=403, 
            detail="Invalid API Key"
        )
    return await call_next(request)




import logging
logger = logging.getLogger(__name__)

@app.middleware("http")     #---------------​详细日志记录​
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time
    
    logger.info(
        f"Method={request.method}, "
        f"Path={request.url.path}, "
        f"Status={response.status_code}, "
        f"Duration={duration:.2f}s"
    )
    return response





@app.middleware("http")     #---------------全局异常捕获​
async def global_exception_handler(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as e:
        logger.error(f"Unhandled exception: {str(e)}")
        return JSONResponse(
            status_code=500,
            content={"message": "Internal server error"}
        )



# ----------------- 性能问题 -----------------
# 性能开销来源:
    # ​​I/O 阻塞操作（主要开销）​​
        # ​​日志写入磁盘​​：同步日志中间件（如 logging 模块）会触发磁盘 I/O，阻塞事件循环，导致请求堆积 。
        # ​​网络调用​​：中间件内同步调用数据库或外部 API（如身份验证中查用户信息）会显著增加延迟 。

    # ​​CPU 密集型操作​​
        # ​​数据压缩​​：全局启用的 GZipMiddleware 压缩大响应体时消耗 CPU，尤其对小响应（<1KB）得不偿失 。
        # ​​复杂验证逻辑​​：深层嵌套的 Pydantic 验证或自定义数据转换占用 CPU 时间 。

    # ​​同步代码阻塞事件循环​​
        # 同步中间件（def 定义）会阻塞整个异步事件循环，即使逻辑简单也会限制并发吞吐量 。
        # 例如：同步中间件中的 time.sleep() 或同步数据库查询会使并发请求串行化处理 。


# 高并发下的性能衰减​​: 当 QPS > 1000 时，中间件开销会被显著放大
# ​​叠加延迟​​：每个中间件增加 1ms，5 个中间件叠加后单请求延迟增加 5ms，万级并发下总延迟放大 50 倍 。
# ​​资源争用​​：同步中间件导致工作进程被占满，异步优势丧失（实测：同步中间件使 8 核服务器 QPS 从 1200 降至 300）。
# 连接池耗尽：中间件频繁创建数据库连接，耗尽连接池后触发等待 。
# ​​内存泄漏风险​​：中间件内未释放资源（如未关闭文件句柄）会导致内存持续增长


# 避免不必要中间件:
    # 禁用非核心功能​​：移除开发调试用的日志中间件、生产环境关闭详细请求日志 。
    # ​​按需启用压缩​​：GZipMiddleware 仅压缩 >1KB 的响应，避免小数据压缩开销
    # app.add_middleware(GZipMiddleware, minimum_size=1024)  # 仅压缩大于1KB的响应
# 异步化改造​​​替换同步库​​：
#     日志 → 改用异步日志库（如 structlog + aiologger）。
#     数据库 → 使用 asyncpg 或 SQLAlchemy 异步引擎 。
#     强制异步声明​​：中间件函数必须用 async def 定义，内部避免同步阻塞调用 。
# ​​功能合并与短路​​
#     ​​合并逻辑​​：将多个中间件合并（如认证+限流），减少调用栈深度 。
#     ​​请求提前终止​​：在中间件内尽早返回无效请求（如认证失败直接返回 401，避免后续开销）
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    if not validate_token(request.headers.get("Authorization")):
        return JSONResponse(status_code=401, content={"detail": "Unauthorized"})  # 认证失败直接返回 401，避免后续开销
    return await call_next(request)

# 性能监控与调优​
# 定位热点​​：使用 Py-Spy 采样 CPU 耗时，识别高开销中间件
py-spy record -o profile.svg -- uvicorn app:app
# 压力测试​​：用 k6 模拟高并发，观察中间件增减对 QPS 的影响 。
# ​​监控告警​​：通过 Prometheus 跟踪中间件平均延迟和错误率 



# 策略：
# 仅验证关键字段，跳过非安全相关字段（如日志标签、辅助元数据）：
# 使用 ijson 流式解析大 JSON，逐块验证避免内存溢出：
# 优先使用 model_validate() 替代 __init__​​
#         model_validate() 直接操作字典输入，跳过多余的类型检查，速度提升 ​​15%​​（但需确保输入为预解析字典）
# 禁用详细错误​​：
#     设置 model_config["return_errors_mode"] = "none"，错误时仅返回 None 而非完整异常，减少 ​​50%​​ 的无效请求处理时间
# ​​Slots 模式减少内存：
# 使用 @dataclass(slots=True) 替代 BaseModel，消除 __dict__ 开销
# 不可变数据结构​​：
# 嵌套列表/字典转换为元组/冻结集合，减少拷贝和哈希开销
class CompactModel(BaseModel):
    data: tuple  # 替代 List
    config: frozenset  # 替代 Dict 

# 递归转迭代：避免栈溢出并提升速度
# 仅验证变更字段，适用于高频局部更新（如配置热重载）


# ​​策略​​	    ​       ​适用场景​​	​               ​性能提升​​	​​              风险与妥协​​
# 部分验证	    非关键字段多的请求体	     20%-40%	            可能跳过安全校验，需审计字段安全性
# model_validate()	预解析数据的高 QPS 接口	10%-15%	            输入必须为字典，失去构造函数灵活性
# Slots 模式	    百万级实例的批量处理	    内存降 70%	        失去动态字段扩展能力
# 流式分块	    >100MB 的 JSON 导入	       避免 OOM	            代码复杂度增加，错误定位困难
# 禁用详细错误	内部高吞吐监控接口	        30%-50%	            调试信息缺失，问题追踪困难


# --------------------------------- 后台任务 -----------------------------------------------
# 响应优先，任务后置​

# 任务注册​​：在路由处理函数中通过 background_tasks.add_task() 注册任务函数及其参数
# 响应返回​​：FastAPI 优先完成 HTTP 响应（如返回 200 OK），用户无需等待任务完成
# ​​任务执行​​：响应发送后，框架按注册顺序执行后台任务
# ​​异步非阻塞​​：任务在事件循环空闲时执行，通过 concurrent.futures 线程池实现（默认使用线程而非协程），避免阻塞主线程
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

def write_log(message: str):
    # 模拟耗时操作（如写入文件）
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

@app.post("/log")
async def submit_log(msg: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, msg)  # 注册任务
    return {"status": "Log queued"}  # 立即返回响应


# 任务队列结构​​​：BackgroundTasks 内部维护一个 ​​先进先出（FIFO）内存队列​​，存储待执行的任务函数及参数
# 执行时机​​：所有注册的任务在响应返回后由 Starlette 框架统一调度，按添加顺序依次执行
# 线程池执行​​：默认使用线程池处理任务（即使任务函数定义为 async def，实际仍由线程池调用）
# ​​并发限制​​：线程池大小受限于 Python 的 ThreadPoolExecutor 默认配置（约 核心数 × 5），高并发时可能成为瓶颈


# 缺陷
# 无持久化​​：任务存储在内存中，若进程崩溃（如服务器重启），未执行的任务将丢失。
# ​​无重试机制​​：任务抛出异常后自动终止，无自动重试支持。
# ​​无状态追踪​​：无法直接查询任务进度或结果


# 轻量级（秒级任务）；日志记录、即时通知；零配置，无需额外组件
# 如：发送欢迎邮件；请求日志写入本地文件；实时性要求低、允许丢失的非关键任务

# 任务耗时是否 < 1分钟  &  不需跨服务或分布式  &  不需任务状态追踪





# ---------------------------------- Pydantic ----------------------------------------------

# ​​字段级验证器 (@field_validator)：针对单个字段的验证，在字段赋值时触发
from pydantic import BaseModel, field_validator  
class Product(BaseModel):  
    price: float  
    @field_validator('price')  
    def price_must_be_positive(cls, v):  
        if v <= 0:  
            raise ValueError('价格必须为正数')  
        return v  # 返回处理后的值  



# 模型级验证器 (@model_validator)：用于跨字段验证或依赖多个字段的逻辑
# 两种模式：mode='before': 在模型初始化前验证原始数据； 'after': 在模型初始化后验证完整对象

class Order(BaseModel):  
    total: float  
    discount: float  
    @model_validator(mode='after')  
    def validate_discount(self):  
        if self.discount > self.total:  
            raise ValueError('折扣金额不能超过订单总额')  
        return self  




# ORM 集成：将 ORM 对象（如 SQLAlchemy、Tortoise-ORM 模型）自动转换为 Pydantic 模型，
#           解决 ORM 对象属性与 Pydantic 字段的映射问题

# Pydantic v1​​：orm_mode = Tru
class UserResponse(BaseModel):
    id: int
    name: str
    class Config:
        orm_mode = True  # 启用 ORM 模式


# ​​Pydantic v2（推荐）​​：from_attributes = True
class UserResponse(BaseModel):
    id: int
    name: str
    model_config = {"from_attributes": True}  # 语义更清晰

# ------------------ 使用场景 --------------------------
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()
class UserORM(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)

# FastAPI 路由
@app.get("/users/{id}", response_model=UserResponse)
def get_user(id: int, db: Session = Depends(get_db)):
    user_orm = db.query(UserORM).get(id)  # 返回 ORM 对象
    return user_orm  # 自动转换为 UserResponse 模型

# 原理​​：启用后，Pydantic 通过 obj.__dict__ 或属性访问（如 getattr）提取字段值，而非默认的字典解析







# alias_generator：自动为所有字段生成别名（如驼峰转蛇形），解决 JSON 字段命名风格与 Python 变量风格的差异
class Item(BaseModel):
    item_name: str
    item_price: float
    
    model_config = {
        "alias_generator": lambda x: x.upper(),  # 生成大写别名（如 ITEM_NAME）
        "populate_by_name": True  # 允许用原始字段名或别名初始化[3,8](@ref)
    }
# ​​实际效果​​：
# 输入 {"ITEM_NAME": "Laptop", "ITEM_PRICE": 999.9} 可自动解析为 Item 实例。
# 输出时默认使用别名（可通过 model_dump(by_alias=False) 禁用）

# extra 控制：额外字段处理策略；控制模型是否允许接收未定义的字段，提升 API 安全性
# extra="allow"	允许未定义字段，存储到 __pydantic_extra__；适用场景：动态数据接收（如元数据）
# extra="ignore"	静默忽略未定义字段；适用场景：默认行为，兼容旧版数据
# extra="forbid"	拒绝未定义字段，抛出 ValidationError；适用场景：严格 API 契约（防客户端错误）

# json_encoders：自定义序列化逻辑；为特定类型（如 datetime、自定义类）定义专属 JSON 序列化方式
    #  复杂类型 通过 json_encoders 转为字符串 → 避免 JSON 序列化错误
# 格式化时间、日期；将数据库枚举类型转为字符串；压缩二进制数据为 Base64
from datetime import datetime
from pydantic import BaseModel

class CustomModel(BaseModel):
    created_at: datetime
    
    model_config = {
        "json_encoders": {
            datetime: lambda dt: dt.strftime("%Y/%m/%d"),  # 自定义时间格式
            "SomeCustomClass": lambda obj: obj.to_json()   # 处理自定义类
        }
    }

obj = CustomModel(created_at=datetime.now())
print(obj.model_dump_json())  # 输出 {"created_at": "2025/07/02"}[6](@ref)

# validate_default=True​​ 对字段默认值执行验证（Pydantic v2 默认启用）。
# ​​arbitrary_types_allowed=True​​ 允许非 Pydantic 类型（如 SQLAlchemy 模型）作为字段类型。
# ​​str_strip_whitespace=True​​ 自动去除字符串字段的首尾空格








# --------------------------------- 循环引用问题，弱引用，动态生成模型 -----------------------------------------------
# user.py
from typing import TYPE_CHECKING, Optional
from pydantic import BaseModel, Field

if TYPE_CHECKING:  # 仅类型检查时导入，避免运行时循环导入
    from .team import Team

class User(BaseModel):
    id: int
    team: Optional["Team"] = Field(default=None, exclude=True)  # 避免序列化循环

# team.py
from pydantic import BaseModel, Field

if TYPE_CHECKING:
    from .user import User

class Team(BaseModel):
    id: int
    members: List["User"] = Field(default_factory=list)

# main.py 显式重建模型
User.model_rebuild()  
Team.model_rebuild()


# TYPE_CHECKING 隔离​​：解决导入依赖
# ​​model_rebuild()​​：在主程序入口统一重建模型，确保所有类型解析完成
# exclude=True​​：防止序列化时陷入循环

# 弱引用（Weak Reference）​​：适用于需避免内存泄漏的场景
import weakref

class Node:
    def __init__(self):
        self.parent = None

class PydanticNode(BaseModel):
    id: str
    parent: Optional[weakref.ReferenceType["PydanticNode"]] = None




# 运行时动态生成模型​​：根据配置或用户输入创建模型，适用于多租户字段自定义、API 网关等场景
from pydantic import create_model, Field

# 根据数据库表结构动态生成模型
fields = {
    "name": (str, Field(..., description="用户名")),
    "age": (Optional[int], Field(default=18, ge=0))
}
DynamicUser = create_model("DynamicUser", **fields)

# 使用动态模型
user_data = {"name": "Alice", "age": 30}
user = DynamicUser(**user_data)  # 自动校验

# 继承
BaseModel = create_model("Base", id=(int, ...))  # 基础字段
ExtendedModel = create_model(
    "Extended",
    __base__=BaseModel,
    role=(str, Field(default="user"))  # 扩展字段
)



# 树形结构​：​	     前向引用 + 深度校验器	
# ​​图状数据环检测：​​	DFS遍历 + 路径跟踪	
# ​​跨模型循环引用​：​	TYPE_CHECKING + model_rebuild() + exclude=True	
# ​​动态模型生成​​：	create_model + 字段映射 + 动态校验器








# ------------------------------------ 路由相关 --------------------------------------------

# FastAPI（基于 Starlette）通过 ​​前缀树（Trie）数据结构​​ 实现高效的 O(1) 路由匹配，
#           其核心原理是将路由路径按片段分割构建树形索引，使查找时间仅与路径深度相关，与路由总数无关
# ​​字典哈希加速 ：总查找时间 = 路径片段数 × O(1)，即 ​​O(K)​​（K 为路径深度），与路由总数无关

# 1. 模块化设计原则​； 按业务拆分路由，避免单文件臃肿；支持独立测试与团队协作开发
# APIRouter核心参数​​：
    # prefix ：统一URL前缀【APIRouter(prefix="/users") → /users/profile】	
    # tags：	OpenAPI文档分组	【tags=["用户管理"] → Swagger UI分类展示】
    # dependencies：	路由器级依赖	【统一认证：dependencies=[Depends(oauth2_scheme)]】
    # responses：	统一响应模型	【responses={404: {"description": "资源不存在"}}】


# 示例：
# 子路由器：orders/routers.py
order_router = APIRouter(prefix="/orders", tags=["订单"])
@order_router.get("/{order_id}")
def get_order(order_id: int): ...

# 父路由器：users/routers.py
from .orders import order_router
user_router = APIRouter(prefix="/users")
user_router.include_router(order_router, prefix="/{user_id}") 
# 生成路径：/users/{user_id}/orders/{order_id}

# 优点：
#     构建层次化API（如用户→订单）
#     复用公共路径参数（如user_id自动传递）


# 路径操作装饰器高级参数​

# ​​数据屏蔽​​：response_model_exclude={"password"} 隐藏敏感字段 
# ​​类型转换​​：response_model=List[UserPublic] 输出简化模型
# 精确状态码：@app.post("/items", status_code=201)
# tags=["认证"] 覆盖路由器分组
# summary="用户登录" 接口摘要（OpenAPI显示）
# ​​操作级覆盖​​：dependencies=[Depends(rate_limit)] 优先级高于路由器级依赖

# 精确响应文档（responses参数）
@app.get(
    "/items/{id}",
    responses={
        200: {"model": Item, "description": "成功获取数据"},
        404: {"description": "物品不存在"},
        403: {"content": {"application/json": {"example": {"error": "权限不足"}}}}
    }
)


# 内置类型扩展​​：FastAPI默认支持类型（如int、UUID），但需处理特殊格式
from pydantic import BaseModel
class CustomDate(BaseModel):  
    @classmethod  
    def __get_validators__(cls):  
        yield cls.validate  
      
    @classmethod  
    def validate(cls, v: str):  
        try:  
            return datetime.strptime(v, "%Y-%m-%d")  
        except ValueError:  
            raise ValueError("日期格式应为YYYY-MM-DD")  

@app.get("/events/{date}")  
def get_events(date: CustomDate):  
    return {"date": date}



# 优先使用枚举类型或自定义转换器，避免无效参数传递


# 静态路径 /users/me 必须定义在动态路径 /users/{id} ​​之前​​，否则后者会优先拦截。
# 使用 priority 参数显式控制优先级
@app.get("/users/me", priority=100)  # 高优先级  
@app.get("/users/{id}", priority=10)   # 低优先级  


# 复杂路由的优化​​
# 正则约束（如 {id:int}）在 Trie 中视为静态节点，不影响查找效率。
# 嵌套路由通过 APIRouter 拆分，每个 Router 独立构建 Trie，避免全局树过深










# -------------------------------------- 高并发下的资源保护 ------------------------------------------
from asyncio import Semaphore  # 信号量控制最大并发数

DB_SEMAPHORE = Semaphore(10)  # 限制10个并发DB操作

@app.get("/limited")
async def limited_query():
    async with DB_SEMAPHORE:  # 超过10请求将等待
        return await db.fetch(...)









# -------------------------------------- ORJSONResponse 代替 json ------------------------------------------
# FastAPI 中的集成方式​
from fastapi.responses import ORJSONResponse  
@app.get("/items/", response_class=ORJSONResponse)  
async def read_items():  
    return [{"item_id": "Foo"}]  

# 全局默认响应类
# app = FastAPI(default_response_class=ORJSONResponse)  











# --------------------------------------------------------------------------------

# 实现文件上传
from fastapi import File, UploadFile
@app.post("/upload-file/")
async def upload_file(file: UploadFile):
    file_content = await file.read()
    with open("new_filename.txt", "wb") as new_file:
        new_file.write(file_content)

    await file.close()








# 在 FastAPI 中，处理会话（Session）和 Cookies 是通过 Starlette 库提供的功能实现的

from fastapi import Response

# 使用 Response 类的 .set_cookie() 方法来设置一个或多个 Cookies。
# 这个方法接受 cookie 名称、值和其他可选参数，如过期时间、路径等。
@app.get("/cookie/")
async def set_cookie(response: Response):
    response.set_cookie(key="my_cookie", value="the_value")



# FastAPI 会自动将请求头中的 “Cookie” 字段解析为字典，并将其作为 Request 对象的 .cookies 属性提供。
# 你可以在路径操作函数中访问此属性来读取请求中的 Cookies。
from fastapi import Request

@app.get("/read-cookie/")
async def read_cookie(request: Request):
    my_cookie = request.cookies.get("my_cookie")
    ...


#  FastAPI 中，没有内置的 Session 支持
# 通过集成第三方库，如 itsdangerous 或 secure-cookie-session 来实现会话管理。


# 虽然 FastAPI 没有直接支持基于 JWT（JSON Web Tokens）的会话管理，
# 但可以结合其他库，如 PyJWT 来实现 JWT 验证和授权。JWT 可以用于无状态的会话管理，
# 在每个请求中携带令牌并在服务器端验证它




# 通过 RedirectResponse 类实现 HTTP 重定向
from fastapi import RedirectResponse

@app.get("/redirect-me/")
async def redirect_me():
    url = "https://example.com/new-location"
    return RedirectResponse(url)
    # 以下是可选：

    # 默认情况下，RedirectResponse 使用 HTTP 状态码 307 (Temporary Redirect)。
    # 如果你想使用其他状态码（如 301, 302），
    # 可以通过在创建 RedirectResponse 时提供额外的 status_code 参数来指定
    # return RedirectResponse(url, status_code=status.HTTP_301_MOVED_PERMANENTLY)

    # 添加额外头部信息
    # headers = {"X-Custom-Header": "Some value"}
    # return RedirectResponse(url, headers=headers)






# 异常：

# 创建自定义异常类：捕获特定类型的错误
class MyCustomError(Exception):
    pass



# 抛出异常
@app.get("/my-endpoint/")
async def my_endpoint():
    if some_condition:
        raise MyCustomError("This is a custom error")  #  或者 ：  ValueError
    ...




# 全局异常处理：@app.exception_handler() 装饰器来定义全局异常处理器
@app.exception_handler(MyCustomError)
async def handle_my_custom_error(request: Request, exc: MyCustomError):
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={"error": str(exc)},
    )




# 路由级别的异常处理：
# 使用 FastAPI.include_router() 函数的 exception_handlers 参数 
# 这个参数应该是一个字典，其中键是异常类，值是处理这些异常的函数。
from fastapi import APIRouter,FastAPI
router = APIRouter()

@router.get("/my-route/")
async def my_route():
    ...

@router.exception_handler(MyCustomError)
async def handle_my_custom_error_for_this_route(request: Request, exc: MyCustomError):
    ...

app = FastAPI()
app.include_router(router, exception_handlers={MyCustomError: handle_my_custom_error_for_this_route})


# 标准 HTTP 异常:  预定义的异常类，如 HTTPException
from fastapi import HTTPException

@app.get("/standard-error/")
async def standard_error():
    raise HTTPException(status_code=404, detail="Not found")


# 除此外还可以使用 python 提供的 异常类
